package com.example.toy.collector;

import com.example.toy.core.record.key.ToyRecordKeyImpl;
import com.example.toy.core.record.key.ToyRecordKey;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.CoderException;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.StructuredCoder;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Arrays;
import java.util.List;

/**
 * BeamToyRecordKeyCoder
 *
 * @author shenbing
 * @date 2021/12/17
 */
public class BeamToyRecordKeyCoder extends StructuredCoder<ToyRecordKey> {
  private final Coder<String> keyCoder = StringUtf8Coder.of();

  private static final BeamToyRecordKeyCoder INSTANCE = new BeamToyRecordKeyCoder();

  public static Coder<ToyRecordKey> of() {
    return INSTANCE;
  }

  @Override
  public void encode(ToyRecordKey value, OutputStream outStream)
      throws CoderException, IOException {
    this.keyCoder.encode(value.getPartitionKey(), outStream);
    this.keyCoder.encode(value.getRawKey(), outStream);
  }

  @Override
  public ToyRecordKey decode(InputStream inStream) throws CoderException, IOException {
    String partitionKey = this.keyCoder.decode(inStream);
    String rawKey = this.keyCoder.decode(inStream);
    return new ToyRecordKeyImpl(partitionKey, rawKey);
  }

  @Override
  public List<? extends Coder<?>> getCoderArguments() {
    return Arrays.asList(this.keyCoder, this.keyCoder);
  }

  @Override
  public void verifyDeterministic() throws NonDeterministicException {}
}
